package com.google.code.garbagecan.cassandrastudy;

import au.com.bytecode.opencsv.CSVReader;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.io.sstable.SSTableSimpleUnsortedWriter;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.List;
import java.util.UUID;

import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
import static org.apache.cassandra.utils.UUIDGen.decompose;

public class CSV2SSTableGenerator {

	public static void main(String[] args) throws Exception {
		CSV2SSTableGenerator generator = new CSV2SSTableGenerator();
		generator.generate("demo", "users", "src/main/resources/users.csv");
	}

	private void generate(String keySpace, String columnFamily, String csvFile) throws IOException {
		File directory = new File(keySpace);
		if (!directory.exists()) {
			directory.mkdir();
		}

		SSTableSimpleUnsortedWriter writer = new SSTableSimpleUnsortedWriter(directory, new RandomPartitioner(), keySpace, columnFamily, AsciiType.instance, null, 64);

		InputStream is = new FileInputStream(new File(csvFile));
		CSVReader csvReader = new CSVReader(new BufferedReader(new InputStreamReader(is)), ',');

		String[] header = csvReader.readNext();
		for (int i = 0, length = header.length; i < length; i++) {
			header[i] = header[i].trim();
			if (i == 0 && header[i].startsWith("#")) {
				header[i] = header[i].replaceFirst("#", "");
			}
		}

		String[] columns = null;
		while ((columns = csvReader.readNext()) != null) {
			if (columns.length != header.length) {
				continue;
			}

			long timestamp = System.currentTimeMillis();
			ByteBuffer uuid = ByteBuffer.wrap(decompose(UUID.randomUUID()));
			writer.newRow(uuid);
			for (int i = 0, length = header.length; i < length; i++) {
				writer.addColumn(bytes(header[i]), bytes(columns[i]), timestamp);
			}
		}

		writer.close();
		is.close();
	}

}
